package org.zachary.imitation.raft.core;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zachary.imitation.raft.NodeId;
import org.zachary.imitation.raft.NodeImpl;
import org.zachary.imitation.raft.PeerId;
import org.zachary.imitation.raft.option.NodeOptions;
import org.zachary.imitation.raft.option.RaftOptions;
import org.zachary.imitation.raft.option.ReplicatorGroupOptions;
import org.zachary.imitation.raft.option.ReplicatorOptions;
import org.zachary.imitation.raft.util.ThreadId;
import org.zachary.utils.lang.Asserts;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;


/**
 * 复制器组的实现类，管理着所有的复制器对象。复制器对象是相对于 Leader 来说的，一个节点一旦成为 Leader，那么该节点就会根据集群配置中的节点的信息，
 * 把其它的 Follower 节点包装成一个个的复制器对象。日志复制就是通过复制器对象实现的。ReplicatorGroupImpl 实例是在 {@link NodeImpl#init(NodeOptions)} 中被创建的。
 */
public class ReplicatorGroupImpl implements ReplicatorGroup {
	private static final Logger logger = LoggerFactory.getLogger(ReplicatorGroupImpl.class);

	/** key = Follower，value = ThreadId 对象。ThreadId 内部就持有着 PeerId 对应的 Follower 节点包装成的复制器对象，每一个复制器对象都是独立的个体，也不会共用线程处理复制器中的回调方法，因为就将每个复制器对象封装到了一个threadId对象中，代表独立的对象 */
	private final ConcurrentMap<PeerId, ThreadId> replicatorMap = new ConcurrentHashMap<>();
	/** 复制器对象需要的配置参数。复制器组下的所有复制器使用相同的配置参数 */
	private ReplicatorOptions replicatorOptions;
	/** raft 框架全局的配置参数 */
	private RaftOptions raftOptions;
	/** 当一个节点成为 Leader 后，经过这个时间之后，就要立刻向 Follower 发送心跳消息了 */
	private long dynamicTimeoutMs = -1;
	/** 选举超时时间 */
	private long electionTimeoutMs = -1;
	/** 用于表明异常节点节点的身份的 Map，相对于 Leader 而言，其它的节点可能是 Follower，也可能是一个 Leaner，该 Map 中存放着其它异常节点的身份映射关系 */
	private final Map<PeerId, ReplicatorType> failureReplicators = new ConcurrentHashMap<>();

	@Override
	public boolean init(final NodeId nodeId, final ReplicatorGroupOptions replicatorGroupOptions) {
		this.dynamicTimeoutMs = replicatorGroupOptions.getHeartbeatTimeoutMs();
		this.electionTimeoutMs = replicatorGroupOptions.getElectionTimeoutMs();
		this.raftOptions = replicatorGroupOptions.getRaftOptions();

		this.replicatorOptions = new ReplicatorOptions();
		this.replicatorOptions.setDynamicHeartBeatTimeoutMs(this.dynamicTimeoutMs);
		this.replicatorOptions.setElectionTimeoutMs(this.electionTimeoutMs);
		// this.options.setRaftRpcService(opts.getRaftRpcClientService());
		this.replicatorOptions.setNode(replicatorGroupOptions.getNode());
		this.replicatorOptions.setTerm(0);
		this.replicatorOptions.setGroupId(nodeId.getGroupId());
		this.replicatorOptions.setServerId(nodeId.getPeerId());
		this.replicatorOptions.setScheduledExecutorService(replicatorGroupOptions.getScheduledExecutorService());
		return true;
	}

	@Override
	public ThreadId getReplicator(final PeerId peer) {
		return this.replicatorMap.get(peer);
	}

	@Override
	public void checkReplicator(PeerId peer, boolean lockNode) {
		// TODO
	}

	/** 复制器组的核心方法，会在 {@link NodeImpl#becomeLeader} 中被调用。当一个节点当选为 Leader 后，会通过该方法把集群中其它节点包装成一个个的复制器对象 */
	@Override
	public boolean addReplicator(final PeerId peer, final ReplicatorType replicatorType, final boolean sync) {
		// ReplicatorGroupImpl 对象是在 NodeImpl#init 中创建的，那个时候 replicatorOption#term 初始化为 0。但是在当前节点当选为 Leader 后，该类对象的 replicatorOption#term 被设置为 Leader 的任期，
		// 然后才执行当前的添加复制器对象的方法，因为每一个复制器对象的任期都要和 Leader 任期一致，所以这里要判断 replicatorOption#term != 0，因为 replicatorOption#term 中的任期接下来要传递给新创建的每一个复制器对象
		Asserts.isTrue(this.replicatorOptions.getTerm() != 0);
		// 添加复制器之前，先从异常节点的 Map 集合中删除一下对应的节点
		this.failureReplicators.remove(peer);
		// 判断复制器对象是否已经添加过了
		if (this.replicatorMap.containsKey(peer)) {
			return true;
		}

		// 得到复制器对象所需要的配置参数
		final ReplicatorOptions opts = this.replicatorOptions == null ? new ReplicatorOptions() : this.replicatorOptions.copy();
		// 设置复制器类型，可能是 Follower，也可能是 Learner
		opts.setReplicatorType(replicatorType);
		opts.setPeerId(peer);

		// TODO sync = false 则检查复制器对应的节点是否可以连接成功。正常不应该是 sync = true 时才检查连通性吗？
		if (!sync) {
//			// 得到RPC客户端
//			final RaftClientService client = opts.getRaftRpcService();
//			// 检查是否能连接成功
//			if (client != null && !client.checkConnection(peer.getEndpoint(), true)) {
//				logger.error("Fail to check replicator connection to peer={}, replicatorType={}.", peer, replicatorType);
//				// 如果无法连接成功，就把当前节点和其身份添加到 failureReplicators Map 中
//				this.failureReplicators.put(peer, replicatorType);
//				return false;
//			}
		}

		// 这里真正创建复制器对象，虽然这里返回了一个 ThreadId 对象，但是该对象内部持有着复制器对象
		final ThreadId rid = Replicator.start(opts, this.raftOptions);
		// 如果创建复制器对象失败则返回 false
		if (rid == null) {
			logger.error("Fail to start replicator to peer={}, replicatorType={}.", peer, replicatorType);
			// 记录失败节点到 failureReplicators Map 中
			this.failureReplicators.put(peer, replicatorType);
			return false;
		}

		// 到这里就一切顺利，把包装着复制器对象的 ThreadId 对象添加到 replicatorMap 中
		return this.replicatorMap.put(peer, rid) == null;
	}

	/**
	 * 重置 {@link ReplicatorOptions#term} 任期的方法，该方法会在 {@link NodeImpl#becomeLeader} 中被调用
	 */
	@Override
	public boolean resetTerm(final long newTerm) {
		if (newTerm <= this.replicatorOptions.getTerm()) {
			return false;
		}
		this.replicatorOptions.setTerm(newTerm);
		return true;
	}


	@Override
	public boolean contains(final PeerId peer) {
		return this.replicatorMap.containsKey(peer);
	}

	@Override
	public void describe(final Printer out) {
		out.print("  replicators: ").println(this.replicatorMap.values());
		out.print("  failureReplicators: ").println(this.failureReplicators);
	}
}
